package com.qs.kafka;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Collections;
import java.util.Properties;

/**
 * 使用kafka-clients构建消费者，消费端消费消息
 *
 * @author QinYupeng
 * @since 2019-12-17 20:20:25
 */
public class ConsumerDemo implements Runnable {

    private final KafkaConsumer kafkaConsumer;

    private final String topic;

    public ConsumerDemo(String topic) {
        this.topic = topic;
        Properties props = new Properties();

        //kafka集群地址
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "172.16.109.130:9092,172.16.109.131:9092,172.16.109.132:9092");

        //定义消费组，一个消费组包含多个consumer，且一个消费组中只能有一个consumer消费topic
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer-group");

        //设置为true，consumer定期的往zookeeper中写入每个分区的offset
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

        //往zookeeper上写offset的频率
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");

        //key的反序列化类型
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        //value的反序列化类型
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        //构建consumer
        kafkaConsumer = new KafkaConsumer<>(props);

        //消费端订阅topic
        kafkaConsumer.subscribe(Collections.singletonList(topic));
    }

    @Override
    public void run() {
        while (true) {
            ConsumerRecords<Integer, String> consumerRecords = kafkaConsumer.poll(1000);
            for (ConsumerRecord<Integer, String> record : consumerRecords) {
                System.out.println("message receive:" + record.value());
                //手动提交
                kafkaConsumer.commitAsync();
            }
        }
    }

    public static void main(String[] args) {
        new Thread(new ConsumerDemo("my-replicated-topic")).start();
    }
}
